package command

import (
	"database/sql"
	"encoding/json"
	"fmt"
	"github.com/parnurzeal/gorequest"
	"log"
	"strconv"
	"time"
)
var (
	tableName  = "spark_executor_metrics"
)


type Executor struct {
	ID               string `json:"id"`
	HostPort         string `json:"hostPort"`
	IsActive         bool   `json:"isActive"`
	RddBlocks        int    `json:"rddBlocks"`
	MemoryUsed       int    `json:"memoryUsed"`
	DiskUsed         int    `json:"diskUsed"`
	TotalCores       int    `json:"totalCores"`
	MaxTasks         int    `json:"maxTasks"`
	ActiveTasks      int    `json:"activeTasks"`
	FailedTasks      int    `json:"failedTasks"`
	CompletedTasks   int    `json:"completedTasks"`
	TotalTasks       int    `json:"totalTasks"`
	TotalDuration    int    `json:"totalDuration"`
	TotalGCTime      int    `json:"totalGCTime"`
	TotalInputBytes  int    `json:"totalInputBytes"`
	TotalShuffleRead int    `json:"totalShuffleRead"`
	TotalShuffleWrite int   `json:"totalShuffleWrite"`
	IsBlacklisted    bool   `json:"isBlacklisted"`
	MaxMemory        int    `json:"maxMemory"`
	AddTime          string `json:"addTime"`
	ExecutorLogs     map[string]interface{} `json:"executorLogs"`
	MemoryMetrics    struct {
		UsedOnHeapStorageMemory    int `json:"usedOnHeapStorageMemory"`
		UsedOffHeapStorageMemory   int `json:"usedOffHeapStorageMemory"`
		TotalOnHeapStorageMemory   int `json:"totalOnHeapStorageMemory"`
		TotalOffHeapStorageMemory  int `json:"totalOffHeapStorageMemory"`
	} `json:"memoryMetrics"`
}

func fetchExecutors(url string) ([]Executor, error) {
	request := gorequest.New()
	// url := "http://localhost:4040/api/v1/applications/local-1688436870722/allexecutors"

	resp, body, errs := request.Get(url).End()
	if len(errs) > 0 {
		return nil, errs[0]
	}
	defer resp.Body.Close()

	var executors []Executor
	err := json.Unmarshal([]byte(body), &executors)
	if err != nil {
		return nil, err
	}

	return executors, nil
}

func saveExecutors(executors []Executor, db *sql.DB, taskID string) error {
	query := fmt.Sprintf("INSERT INTO %s (task_id, collect_time, id, host_port, is_active, rdd_blocks, " +
		"memory_used, disk_used, total_cores, max_tasks, active_tasks, " +
		"failed_tasks, completed_tasks, total_tasks, total_duration, total_gc_time, " +
		"total_input_bytes, total_shuffle_read, total_shuffle_write, is_blacklisted, " +
		"max_memory, add_time, executor_logs, used_on_heap_storage_memory, used_off_heap_storage_memory, " +
		"total_on_heap_storage_memory, total_off_heap_storage_memory) " +
		"VALUES " +
		"(?, ?, ?, ?, ?, ?, " +
		"?, ?, ?, ?, ?, " +
		"?, ?, ?, ?, ?, " +
		"?, ?, ?, ?, " +
		"?, ?, ?, ?, ?, " +
		"?, ?)", tableName)
	log.Println(query)
	stmt, err := db.Prepare(query)
	if err != nil {
		fmt.Println(err)
		return err
	}
	defer stmt.Close()

	for _, executor := range executors {
		// taskID := "local-1688436870722" // 任务ID
		collectTime := time.Now()      // 采集时间

		executorJSON, err := json.Marshal(executor)
		if err != nil {
			log.Println("Failed to marshal executor:", err)
			continue
		}

		_, err = stmt.Exec(taskID, collectTime, executor.ID, executor.HostPort, strconv.FormatBool(executor.IsActive), executor.RddBlocks,
			executor.MemoryUsed, executor.DiskUsed, executor.TotalCores, executor.MaxTasks, executor.ActiveTasks,
			executor.FailedTasks, executor.CompletedTasks, executor.TotalTasks, executor.TotalDuration, executor.TotalGCTime,
			executor.TotalInputBytes, executor.TotalShuffleRead, executor.TotalShuffleWrite, strconv.FormatBool(executor.IsBlacklisted),
			executor.MaxMemory, ParseTime(executor.AddTime), string(executorJSON), executor.MemoryMetrics.UsedOnHeapStorageMemory,
			executor.MemoryMetrics.UsedOffHeapStorageMemory, executor.MemoryMetrics.TotalOnHeapStorageMemory, executor.MemoryMetrics.TotalOffHeapStorageMemory)
		if err != nil {
			log.Println("Failed to execute query:", err)
		}
	}

	return nil
}

func ParseTime(str string) time.Time {
	// str := "2023-07-04T02:14:30.746GMT"
	layout := "2006-01-02T15:04:05.000Z0700" // 定义时间字符串的格式
	// fmt.Println(str)
	// 将 "GMT" 替换为 "+0000"
	str = str[:len(str)-3] + "+0000"
	t, err := time.Parse(layout, str)
	if err != nil {
		// 发生错误，处理错误情况
		log.Println("转换失败:", err)
	} else {
		// 转换成功，使用 t 进行后续操作
		// fmt.Println("转换结果:", t)
	}
	return t
}

func ExecuteSparkCommand(url , executorID string, db *sql.DB) {
	executors, err := fetchExecutors(url)
	if err != nil {
		log.Println("Failed to fetch executors:", err)
		return
	}

	err = saveExecutors(executors, db, executorID)
	if err != nil {
		log.Println("Failed to save executors:", err)
	}
}